package com.foogui.demo.consume;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.stereotype.Service;

import java.util.List;

@Slf4j
//@Service
@RocketMQMessageListener(consumerGroup = "consume_group", topic = "OrderTopic")
public class SequenceConsumer {

    public static void main(String[] args) throws Exception {
        //1.创建消费者Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consume_group");
        //2.指定Nameserver地址
        consumer.setNamesrvAddr("localhost:9876");
        //3.订阅主题Topic和Tag
        consumer.subscribe("OrderTopic", "*");

        /**
         * 设置第一次启动consumer从队列头消费
         * 如果不是第一次启动，name按照上次的消费位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + ", msg=" + new String(msg.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }

        });
        //5.启动消费者consumer
        consumer.start();
    }

}
